﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Diagnostics;
using GeneralDFS.DataEntity.Enum;
using GeneralDFS.DataEntity.IO;

namespace GeneralDFS.DataEntity.Nodes
{
    /// <summary>
    /// 文件块元数据和节点列表
    /// </summary>
    [Serializable]
    public class BlockTargetPair
    {
        public readonly FileBlockMetaData BlockMetaData;
        public readonly DataNodeInfoWithBlocks[] Targets;

        public BlockTargetPair(FileBlockMetaData blockMetaData, DataNodeInfoWithBlocks[] targets)
        {
            BlockMetaData = blockMetaData;
            Targets = targets;
        }
    }

    [Serializable]
    public class DataNodeInfoWithBlocks:DataNodeID,IComparable<DataNodeInfoWithBlocks>
    {
        private volatile FileBlockInfo _blockList;
        private int _numBlocks = 0;

        public long FreeSpace { set; get; }

        public long UsedSpace { set; get; }

        public DataNodeState State { set; get; }

        public long LastUpdateTime { set; get; }

        public Boolean IsAlive = false;
// ReSharper disable ConvertToConstant.Local
// ReSharper disable FieldCanBeMadeReadOnly.Local
        private Boolean _firstBlockReport = true;
// ReSharper restore FieldCanBeMadeReadOnly.Local
// ReSharper restore ConvertToConstant.Local

        private readonly FileBlockQueue<BlockTargetPair> _replicateBlocks=new FileBlockQueue<BlockTargetPair>();

        /// <summary>
        /// When set to true, the node is not in include list and is not allowed
        /// to communicate with the namenode
        /// </summary>
        private Boolean _disallowed = false;

        public DataNodeInfoWithBlocks(int replication)
        {
            _blockList=new FileBlockInfo(replication);
        }

        public Boolean AddBlock(FileBlockInfo blockInfo)
        {
            if (!blockInfo.AddNode(this)) return false;

            _blockList = blockInfo.ListInsert(_blockList, this);
            _numBlocks++;
            return true;
        }

        public Boolean RemoveBlock(FileBlockInfo blockInfo)
        {
            _blockList = blockInfo.ListRemove(_blockList, this);
            if (blockInfo.RemoveNode(this))
            {
                _numBlocks--;
                return true;
            }
            return false;
        }

        public FileBlockInfo GetFileBlockList()
        {
            return _blockList;
        }

        public int MoveBlockToHead(FileBlockInfo blockInfo, int curIndex, int headIndex)
        {
            _blockList = blockInfo.MoveBlockToHead(_blockList, this, curIndex, headIndex);
            return curIndex;
        }

        public int NumCount()
        {
            return _numBlocks;
        }

        public FileBlockInfo ReplaceBlock(FileBlockInfo oldBlock, FileBlockInfo newBlock)
        {
            if (RemoveBlock(oldBlock))
            {
                AddBlock(newBlock);
            }
            return newBlock;
        }

        public void AddBlockToBeReplicated(FileBlockInfo blockInfo, DataNodeInfoWithBlocks[] dataNodeInfoWithBlockses)
        {
            Debug.Assert(blockInfo != null && dataNodeInfoWithBlockses != null && dataNodeInfoWithBlockses.Length>0);
            _replicateBlocks.Enqueue(new BlockTargetPair(blockInfo, dataNodeInfoWithBlockses));
        }

        public int GetNumberOfBlocksToBeReplicated()
        {
            return _replicateBlocks.Size();
        }

        public List<BlockTargetPair> GetReplicationCommand(int maxTransfers)
        {
            return _replicateBlocks.Poll(maxTransfers);
        }

        public Boolean IsFirstBlockReport()
        {
            return _firstBlockReport;
        }

        public int CompareTo(DataNodeInfoWithBlocks other)
        {
            return base.CompareTo(other);
        }
    }
}
